/**
 * 
 */
package com.etlmaster.executor.steps.extract;

import java.sql.Connection;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

import freemarker.core.FreeMarkerTemplate;

/**
 * @author FengL
 * 
 */
@StepInstruction("hive2Hdfs")
public class Hive2HdfsStep extends Step {

	private Hive2HdfsStepMeta meta;

	/**
	 * @param meta
	 */
	public Hive2HdfsStep(StepMeta meta) {
		super(meta);
		this.meta = (Hive2HdfsStepMeta) meta;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.asiainfo.dacp.dp.executor.Step#handle()
	 */
	@Override
	protected int handle() {
		int flag = StepState.RUNNING_SUCCESSED;

		String dsType = DataSourceHolder.getInstance().getDataSourceType(
				meta.getDsName());
		if (!dsType.equalsIgnoreCase("hive")) {
			meta.log("WARN", "选择非hive数据库({})", meta.getDsName());
			return StepState.RUNNING_FAILED;
		}
		DataSourceDefine def = DataSourceHolder.getInstance()
				.getDataSourceDefine(meta.getDsName());

		if (def != null) {

		} else {
			meta.log("ERROR", "未分配数据库资源");
			return StepState.RUNNING_FAILED;
		}

		Map<String, String> data = new HashMap<String, String>();
		data.put("isLocal", meta.getIsLocal());
		data.put("targetDir", meta.getTargetDir());
		data.put("filedSpit", meta.getFieldSplit());
		data.put("sql", meta.getSql());
		data.put("mapNum", meta.getMapNum());
		String extSql = FreeMarkerTemplate.getFpl("Hive2Hdfs", data);

		try {
			JdbcTemplate jdbcTemplate = null;

			jdbcTemplate = new JdbcTemplate(meta.getDsName());
			Connection conn = jdbcTemplate.getDataSource().getConnection();
			Statement stmt = conn.createStatement();

			if (extSql != null && extSql.length() > 0) {
				// String hiveSql=StringUtils.replaceSql(meta.getSql());

				String hiveSql = extSql;
				String[] sqls = hiveSql.split("(?<!\\\\);");
				for (String sql : sqls) {
					sql = sql.trim();
					if (sql.length() > 0) {
						// sql = meta.variableSubstitution(sql.replace("\\;",
						// ";"));
						sql = sql.trim();
						meta.log("INFO", "begin:dbname:{} sql:{}",
								meta.getDsName(), sql);

						stmt.execute(sql);
						meta.log("INFO", "成功行数：{},数据源为:{}", 0 + "",
								meta.getDsName());
					}
				}
			} else {
				meta.log("ERROR", "配置的SQL语句为空，无法执行.或则部分变量未声明，请检查！");
				flag = StepState.RUNNING_FAILED;
			}
		} catch (Exception e) {
			meta.log("ERROR", e.getMessage());
			flag = StepState.RUNNING_FAILED;
		}
		return flag;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.asiainfo.dacp.dp.executor.Step#stop()
	 */
	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
